package com.viu.technology.config.rabbit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.ExchangeBuilder;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.core.QueueBuilder;
import org.springframework.amqp.core.TopicExchange;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.support.RetryTemplate;

import com.viu.technology.constant.RabbitQueueEnum;

@Configuration
@EnableRabbit
public class RabbitConfiguration {

	private static Logger log = LoggerFactory.getLogger(RabbitConfiguration.class);

	@Bean
	public ConnectionFactory factory() {
		CachingConnectionFactory factory = new CachingConnectionFactory();
		factory.setHost("127.0.0.1");
		factory.setPort(5672);
		factory.setUsername("root");
		factory.setPassword("root");

		// 属性分别设置为“true”，支持确认和返回的消息
		factory.setPublisherConfirms(true);
		factory.setPublisherReturns(true);

		// 最多66条信道，超过着等待前面任务完成或超时
		factory.setChannelCacheSize(66);
		return factory;
	}

	@Bean
	public AmqpAdmin amqpAdmin() {
		RabbitAdmin admin = new RabbitAdmin(factory());
		// 可以在此处做队列的初始化工作
		// admin.declareExchange(new TopicExchange("topic.exchange.test"));
		// admin.declareQueue(new Queue("spring.boot.rabbit.test.send"));
		return admin;
	}

	/** 因为要设置回调类，所以应是prototype类型，如果是singleton类型，则回调类为最后一次设置 */
	@Bean
	@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
	public RabbitTemplate rabbitTemplate() throws Throwable {
		RabbitTemplate rabbitTemplate = new RabbitTemplate(factory());

		// 初始化重试策略
		RetryTemplate retryTemplate = new RetryTemplate();

		ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
		backOffPolicy.setInitialInterval(500); // 第一次重试间隔
		backOffPolicy.setMultiplier(10.0); // 每次重试时间间隔
		backOffPolicy.setMaxInterval(10000); // 最大重试时间间隔

		retryTemplate.setBackOffPolicy(backOffPolicy);
		rabbitTemplate.setRetryTemplate(retryTemplate);
		rabbitTemplate.setMandatory(true);

		// 设置返回消息函数setPublisherConfirms必须为true，一个RabbitTemplate只能有一个
		/*
		 * rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
		 * 
		 * @Override public void returnedMessage(Message message, int replyCode, String
		 * replyText, String exchange, String routingKey) { log.info("正在返回消息"); } });
		 */
		// 确认消息是否到达broker服务器，也就是只确认是否正确到达exchange中即可，只要正确的到达exchange中，broker即可确认该消息返回给客户端ack。
		rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
			@Override
			public void confirm(CorrelationData correlationData, boolean ack, String cause) {
				if (ack) {
					log.debug("消息发送成功");
				} else {
					log.debug("消息发送失败,id: {} , 原因: {}", correlationData.getId(), cause);
				}
			}
		});
		return rabbitTemplate;
	}

	/*
	 * ======================================= 测试队列
	 * ==================================
	 */

	@Bean
	public Queue testQueue() {
		return new Queue("queue.rabbit.test");
	}

	@Bean
	public TopicExchange topicExchange() {
		return new TopicExchange("exchange.rabbit.test");
	}

	@Bean
	public Binding testBinding(Queue chatDurableQueue, DirectExchange chatDurableExchange) {
		return BindingBuilder.bind(chatDurableQueue).to(chatDurableExchange).with("routingKey.rabbit.test");
	}

	/*
	 * ======================================= 死信队列Demo
	 * ==================================
	 */

	@Bean
	public DirectExchange expireExchange() {
		return (DirectExchange) ExchangeBuilder.directExchange("exchange.ttl.expire").durable(true).build();
	}

	@Bean
	public Queue expireQueue() {
		return QueueBuilder.durable("queue.ttl.expire") // 配置到期后转发的交换
				.withArgument("x-dead-letter-exchange", "exchange.ttl.message") // 配置到期后转发的路由键
				.withArgument("x-dead-letter-routing-key", "routingKey.ttl.message").build();
	}

	@Bean
	public DirectExchange messageExchange() {
		return (DirectExchange) ExchangeBuilder.directExchange("exchange.ttl.message").durable(true).build();
	}

	@Bean
	public Queue messageQueue() {
		return new Queue("queue.ttl.message");
	}

	@Bean
	public Binding messageBinding(DirectExchange messageExchange, Queue messageQueue) {
		return BindingBuilder.bind(messageQueue).to(messageExchange).with("routingKey.ttl.message");
	}

	@Bean
	public Binding messageTtlBinding(Queue expireQueue, DirectExchange expireExchange) {
		return BindingBuilder.bind(expireQueue).to(expireExchange).with("routingKey.ttl.expire");
	}

	/*
	 * =============================================================================
	 * ===========
	 */

	/*
	 * ====================================消息持久化队列声明
	 * ==================================
	 */

	@Bean
	public Queue chatDurableQueue() {
		return new Queue(RabbitQueueEnum.CHAT_MESSAGE_DURABLE.getQueueName());
	}

	@Bean
	public DirectExchange chatDurableExchange() {
		return new DirectExchange(RabbitQueueEnum.CHAT_MESSAGE_DURABLE.getExchangeName());
	}

	@Bean
	public Binding chatDurableBinding(Queue chatDurableQueue, DirectExchange chatDurableExchange) {
		return BindingBuilder.bind(chatDurableQueue).to(chatDurableExchange)
				.with(RabbitQueueEnum.CHAT_MESSAGE_DURABLE.getRoutingKey());
	}
	/*
	 * =============================================================================
	 * ============
	 */

}
